import sys

from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtGui import QTextCursor
from kafka import KafkaConsumer

from PyQt5 import QtWidgets


class ProducerGui(QtWidgets.QWidget):
    """KAFKA消费者"""

    def __init__(self):
        super().__init__()

        self.setWindowTitle("KAFKA消费者模拟工具")

        self.host = QtWidgets.QLineEdit(self)
        self.host.setPlaceholderText("KAFKA服务器地址")

        self.port = QtWidgets.QLineEdit(self)
        self.port.setPlaceholderText("端口")

        self.topic = QtWidgets.QLineEdit(self)
        self.topic.setPlaceholderText("TOPIC")

        self.connect_button = QtWidgets.QPushButton("连接", self)
        self.close_button = QtWidgets.QPushButton("清空接收消息", self)

        self.receive_msg_edit = QtWidgets.QTextEdit(self)
        self.receive_msg_edit.setReadOnly(True)

        layout1 = QtWidgets.QHBoxLayout()
        layout1.addWidget(QtWidgets.QLabel("地址："))
        layout1.addWidget(self.host)
        layout1.addWidget(QtWidgets.QLabel("端口："))
        layout1.addWidget(self.port)
        layout1.addWidget(QtWidgets.QLabel("TOPIC："))
        layout1.addWidget(self.topic)
        layout1.addWidget(self.connect_button)
        layout1.addWidget(self.close_button)

        layout2 = QtWidgets.QVBoxLayout()
        layout2.addWidget(QtWidgets.QLabel("接收到的消息："))
        layout2.addWidget(self.receive_msg_edit)

        layout3 = QtWidgets.QVBoxLayout()
        layout3.addLayout(layout1)
        layout3.addLayout(layout2)
        self.setLayout(layout3)

        self.ui_init()

        self.connect_button.clicked.connect(self.connect_to_kafka)
        self.close_button.clicked.connect(self.clear_received_messages)

    def connect_to_kafka(self):
        """连接到kafka"""
        host = self.host.text()
        port = self.port.text()
        topic = self.topic.text()
        print(host, port, topic)
        self.consumer_thread = KafkaConsumerThread(f"{host}:{port}", topic)
        self.consumer_thread.message_received.connect(self.handle_message_received)
        self.consumer_thread.start()

    def handle_message_received(self, message):
        self.receive_msg_edit.insertPlainText(message + "\n\n")
        self.receive_msg_edit.moveCursor(QTextCursor.End)
        self.receive_msg_edit.ensureCursorVisible()

    def clear_received_messages(self):
        self.receive_msg_edit.clear()

    def ui_init(self):
        self.host.setText("192.168.120.221")
        self.port.setText("9094")
        self.topic.setText("SYSTEM_MASTER_IN")


class KafkaConsumerThread(QThread):
    message_received = pyqtSignal(str)

    def __init__(self, kafka_url, topic):
        super().__init__()
        self.kafka_url = kafka_url
        self.topic = topic
        self.consumer = None

    def run(self):
        self.consumer = KafkaConsumer(
            self.topic,
            bootstrap_servers=self.kafka_url,
            auto_offset_reset="earliest",
            enable_auto_commit=True,
        )
        for message in self.consumer:
            self.message_received.emit(message.value.decode("utf-8"))


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    window = ProducerGui()
    window.show()
    sys.exit(app.exec_())
